package cn.yitun.chat.yitunservermaincoresingle.main.business.chat.mq;

import cn.yitun.chat.yitunservercommonlib.dataentity.data.base.chat.ChatItem;
import cn.yitun.chat.yitunservercommonlib.dataentity.data.message.chat.Content;
import cn.yitun.chat.yitunserverimapicore.business.chat.service.UserChatHistoryService;
import cn.yitun.chat.yitunserverimapicore.mq.UserChatQueue;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * Description <br>
 * Date 2019-06-28 14:41:54<br>
 * 
 * @author yitun
 * @since 1.0.0
 */
@Component
public class UserChatQueueImpl implements UserChatQueue {

	@Resource
	UserChatHistoryService userChatHistoryService;

	UserChatEventProducer producer;

	public UserChatQueueImpl() {
		UserChatEventFactory factory = new UserChatEventFactory();
		int bufferSize = 1024 * 1024;
		Disruptor<UserChatEvent> disruptor = new Disruptor<>(factory, bufferSize, DaemonThreadFactory.INSTANCE);
		disruptor.handleEventsWith((event, sequence, endOfBatch) -> on(event));
		disruptor.start();
		RingBuffer<UserChatEvent> ringBuffer = disruptor.getRingBuffer();
		producer = new UserChatEventProducer(ringBuffer);
	}

	@Override
	public void add(Long sendUserId, Long receiveUserId, Content content, List<ChatItem> chatItems, boolean send) {
		producer.onData(sendUserId, receiveUserId, content, chatItems, send);
	}

	public void on(UserChatEvent e) {
		Content c = e.getContent();
		userChatHistoryService.addUserChatHistory(e.isSend(), e.getSendUserId(), e.getReceiveUserId(), c.getKey(), e.getChatItems());
	}
}
